#pragma once
#include <string>
#include "../common/message.hpp"
#include "../common/net.hpp"
#include <unordered_set>
#include<vector>
namespace rpc
{
    namespace server
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            TopicManager() {}
            void onTopicRequestor( const BaseConnection::ptr &conn, TopicRequest::ptr& topic)
            {
                TOPIC_OPTION type = topic->getTopicOption();
                bool ret = true;
                switch (type)
                {
                    // 主题的创建
                case TOPIC_OPTION::TOPIC_CREATE:
                    createTopic(conn, topic);
                    break;
                // 主题的删除
                case TOPIC_OPTION::TOPIC_REMOVE:
                    removeTopic(conn, topic);
                    break;
                // 主题的取消订阅
                case TOPIC_OPTION::TOPIC_CANCEL:
                    topicCancel(conn, topic);
                    break;
                // 主题的发布
                case TOPIC_OPTION::TOPIC_PUBLISH:
                    ret = topicPublish(conn, topic);
                    break;
                // 主题的订阅
                case TOPIC_OPTION::TOPIC_SUBSCRIBE:
                    ret = topicSubscribe(conn, topic);
                    break;
                default:
                    errResponse(conn, topic, topic->getTopicOption());
                }

                if(!ret)
                {
                    return errResponse(conn, topic, topic->getTopicOption());
                }

                return topicResponse(conn, topic);
            }
            // 订阅者断开连接时要删除它的连接信息
            void onShutDown(const BaseConnection::ptr &conn) 
            {
                //消息发布者关闭连接不需要做任何操作
                //消息订阅者关闭连接时要删除她所关联的数据
                SubScriber::ptr subscriber;
                std::vector<Topic::ptr> topics;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto subscriber = _subscribers.find(conn);
                    if( subscriber== _subscribers.end())
                    {
                        return;
                    }
                    for(auto topic_name:subscriber->second->_topics )
                    {
                        auto it = _topics.find(topic_name);
                        if(it == _topics.end())
                        {
                            continue;
                        }
                        topics.push_back(it->second);
                    }

                    _subscribers.erase(subscriber);
                }

                for(auto topic:topics)
                {
                    topic->removeSubscribe(subscriber);
                }
            }

        private:
            // 错误响应
            void errResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg, TOPIC_OPTION type)
            {
                auto resp = MessageFactory::create<TopicResponse>();
                resp->setId(msg->getId());
                resp->setRecode(MY_CODE::RCODE_INVAILED_OPTYPE);
                resp->setMytype(MY_TYPE::RSP_TOPIC);
                conn->send(resp);
            }

            //主题响应
            void topicResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                auto resp = MessageFactory::create<TopicResponse>();
                resp->setId(msg->getId());
                resp->setMytype(MY_TYPE::RSP_TOPIC);
                resp->setRecode(MY_CODE::RCODE_OK);
               
                conn->send(resp);
            }
            // 主题的创建
            void createTopic(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                std::string tName = msg->getKeyTopic();
                auto _topic = std::make_shared<Topic>(tName);
                _topics.insert(std::make_pair(tName, _topic));
            }

            // 主题的删除
            void removeTopic(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                std::string tName = msg->getKeyTopic();
                std::unordered_set<SubScriber::ptr> subScriber;
                {
                    // 先找到受到主题删除影响的连接并集合在一起
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _topics.find(tName);
                    if (it == _topics.end())
                    {
                        return;
                    }
                    // 根据主题在订阅者中找到订阅者
                    subScriber = it->second->_substribe;
                    _topics.erase(it);
                }

                for (auto &sub_it : subScriber)
                {
                    sub_it->removeTopic(tName);
                }
            }

            // 主题的订阅
            bool topicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 1，先找到主题对象和订阅对象
                Topic::ptr topic;
                SubScriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->getKeyTopic());
                    if (topic_it == _topics.end())
                    {
                        errlog("主题未找到");
                        return false;
                    }
                    else
                    {
                        topic = topic_it->second;
                    }

                    // 根据连接找订阅者
                    auto subscriber_it = _subscribers.find(conn);
                    if (subscriber_it == _subscribers.end())
                    {
                        // 创建订阅者
                        subscriber = std::make_shared<SubScriber>(conn);
                        _subscribers.insert(std::make_pair(conn, subscriber));
                    }
                    else
                    {
                        subscriber = subscriber_it->second;
                    }
                }

                // 在主题中新增一个订阅对象的连接，在订阅者中增加一个主题
                topic->addSubscribe(subscriber);
                subscriber->appendTopic(topic->getTopicName());
                return true;
            }

            // 订阅的取消
            void topicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 1，先找到主题对象和订阅对象
                Topic::ptr topic;
                SubScriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->getKeyTopic());
                    if (topic_it != _topics.end())
                    {
                        topic = topic_it->second;
                    }

                    // 根据连接找订阅者
                    auto subscriber_it = _subscribers.find(conn);
                    if (subscriber_it != _subscribers.end())
                    {
                        subscriber = subscriber_it->second;
                    }
                }

                if (topic)
                {
                    topic->addSubscribe(subscriber);
                }

                if (subscriber)
                {
                    subscriber->removeTopic(msg->getKeyTopic());
                }
            }

            // 主题的消息发布
            bool topicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                Topic::ptr topic;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _topics.find(msg->getKeyTopic());
                    if (it == _topics.end())
                    {
                        return false;
                    }

                    topic = it->second;
                }

                topic->PushTopic(msg);
                return true;
            }

        private:
            struct SubScriber
            {
                using ptr = std::shared_ptr<SubScriber>;
                std::mutex _mutex;
                BaseConnection::ptr _conn;
                std::unordered_set<std::string> _topics; // 订阅者订阅的主题
                SubScriber(const BaseConnection::ptr &conn)
                    : _conn(conn)
                {
                }
                void appendTopic(const std::string &topic)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _topics.insert(topic);
                }
                void removeTopic( const std::string &topic)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _topics.erase(topic);
                }
            };

            struct Topic
            {
                using ptr = std::shared_ptr<Topic>;
                std::mutex _mutex;
                Topic(const std::string &topic_name)
                    : _topic_name(topic_name)
                {
                }
                std::string getTopicName()
                {
                    return _topic_name;
                }
                // 添加消息订阅者
                void addSubscribe(const SubScriber::ptr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _substribe.insert(subscriber);
                }
                // 发送消息订阅者
                void removeSubscribe(const SubScriber::ptr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _substribe.erase(subscriber);
                }
                // 根据连接来推送消息
                void PushTopic(const TopicRequest::ptr &msg)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    for (auto &subscriber : _substribe)
                    {
                        subscriber->_conn->send(msg);
                    }
                }

                std::string _topic_name;
                std::unordered_set<SubScriber::ptr> _substribe; // 当前主题的订阅者
            };

        private:
            std::mutex _mutex;
            std::unordered_map<std::string, Topic::ptr> _topics;                   // 将名字与主题管理起来
            std::unordered_map<BaseConnection::ptr, SubScriber::ptr> _subscribers; // 将连接与发布者管理起来
        };
    }
}